#include "app_mqtt.h"

#define SERVER_URL "tcp://192.168.56.228:1883"
#define CLIENT_ID "my_client"
#define TOPIC_PULL "pull"
#define TOPIC_PUSH "push"

typedef struct
{
    MQTTClient client;
    void (*recv_cb)(void *data, int len);
} app_mqtt_t;

static app_mqtt_t my_mqtt;

void conn_lost_cb(void *context, char *cause);
int msg_arrived_cb(void *context, char *topicName, int topicLen, MQTTClient_message *message);
void delivery_complete_cb(void *context, MQTTClient_deliveryToken dt);

gate_state_t app_mqtt_init(void)
{

    // 1. 创建mqtt客护端
    if (MQTTClient_create(&my_mqtt.client,
                          SERVER_URL,
                          CLIENT_ID,
                          MQTTCLIENT_PERSISTENCE_NONE,
                          NULL) != MQTTCLIENT_SUCCESS)
    {
        log_error("MQTTClient_create failed");
        return GATE_ERROR;
    }
    log_info("MQTTClient_create success");
    // 2. 设置回调函数
    if (MQTTClient_setCallbacks(my_mqtt.client,
                                NULL,
                                conn_lost_cb,
                                msg_arrived_cb,
                                delivery_complete_cb) != MQTTCLIENT_SUCCESS)
    {
        log_error("MQTTClient_setCallbacks failed");

        return GATE_ERROR;
    }
    log_info("MQTTClient_setCallbacks success");
    // 3. 连接服务器
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    if (MQTTClient_connect(my_mqtt.client, &conn_opts) != MQTTCLIENT_SUCCESS)
    {
        log_error("MQTTClient_connect failed");
        return GATE_ERROR;
    }
    log_info("MQTTClient_connect success");

    // 4. 订阅主题
    if (MQTTClient_subscribe(my_mqtt.client, TOPIC_PULL, 0) != MQTTCLIENT_SUCCESS)
    {
        log_error("MQTTClient_subscribe failed");
        return GATE_ERROR;
    }
    log_info("MQTTClient_subscribe success: %s", TOPIC_PULL);

    return GATE_OK;
}

void app_mqtt_deinit(void)
{
    // 取消主题订阅
    MQTTClient_unsubscribe(my_mqtt.client, TOPIC_PULL);
    // 断开连接
    MQTTClient_disconnect(my_mqtt.client, 10000);
    // 销毁mqtt客户端,释放内存
    MQTTClient_destroy(&my_mqtt.client);
}

void conn_lost_cb(void *context, char *cause)
{
    log_error("Connection lost: %s", cause);
    // 重新连接
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    while (1)
    {
        if (MQTTClient_connect(my_mqtt.client, &conn_opts) != MQTTCLIENT_SUCCESS)
        {
            log_error("MQTTClient_connect failed");
            sleep(3);
            continue;
        }
        log_info("MQTTClient_connect success");
        break;
    }

    // 重新订阅主题
    while (1)
    {
        if (MQTTClient_subscribe(my_mqtt.client, TOPIC_PULL, 0) != MQTTCLIENT_SUCCESS)
        {
            log_error("MQTTClient_subscribe failed");
            sleep(3);
            continue;
        }
        log_info("MQTTClient_subscribe success: %s", TOPIC_PULL);
        break;
    }
}
int msg_arrived_cb(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    // log_info("Message arrived: %.*s", message->payloadlen, message->payload);
    if (my_mqtt.recv_cb)
    {
        my_mqtt.recv_cb(message->payload, message->payloadlen);
    }

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}
void delivery_complete_cb(void *context, MQTTClient_deliveryToken dt)
{
    log_info("Message delivered: %d", dt);
}

gate_state_t app_mqtt_send(void *data, int len)
{
    if (MQTTClient_publish(my_mqtt.client, TOPIC_PUSH, len, data, 0, 0, NULL) != MQTTCLIENT_SUCCESS)
    {
        log_error("MQTTClient_publish failed");
        return GATE_ERROR;
    }
    log_info("MQTTClient_publish success");
    return GATE_OK;
}

void app_mqtt_resister_revc_cb(void (*recv_cb)(void *data, int len))
{
    my_mqtt.recv_cb = recv_cb;
}
